package com.lvmama.rhino.service.impl;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.QueryStringQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket;
import org.elasticsearch.search.aggregations.bucket.terms.TermsBuilder;
import org.elasticsearch.search.aggregations.metrics.sum.InternalSum;
import org.elasticsearch.search.aggregations.metrics.sum.SumBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.InitializingBean;

import com.lvmama.rhino.utils.Constants;  
  
public class ElasticsearchDataService implements InitializingBean {  
  
    private static final Logger logger = Logger.getLogger(ElasticsearchDataService.class);  
  
   /* @Autowired  
    private Client client; */ 
  
    //private String esIndexName = "rhino"; 
    private String esIndexName = Constants.H5Page_INDEX_NAME; 
    
    public TransportClient getClient() {
        Settings settings = Settings.settingsBuilder().put("cluster.name", Constants.getConfig("elasticsearch.cluster.name")).build();
        TransportClient client = null;
        try {
            String[] ips = Constants.getConfig("elasticsearch.host").split(",");
            int port = Integer.parseInt(Constants.getConfig("elasticsearch.port"));
            client = TransportClient.builder().settings(settings).build();
            for(String ip : ips){   
                client = client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ip), port));
            }
            //client = TransportClient.builder().settings(settings).build().addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(Constants.getConfig("elasticsearch.host")), 9300));
        }
        catch (UnknownHostException e) {
            e.printStackTrace();
        }
       return client;
    }
  
    /** 查询 id */  
    public List<String> queryId(String type, String[] fields, String content,  
            String sortField, SortOrder order, int from, int size) {  
        SearchRequestBuilder reqBuilder = getClient().prepareSearch(esIndexName)  
                .setTypes(type).setSearchType(SearchType.DEFAULT)  
                .setExplain(true);  
        QueryStringQueryBuilder queryString = QueryBuilders.queryStringQuery("\""  
                + content + "\"");  
        for (String k : fields) {  
            queryString.field(k);  
        }  
        queryString.minimumShouldMatch("10");  
        reqBuilder.setQuery(QueryBuilders.boolQuery().should(queryString))  
                .setExplain(true);  
        if (StringUtils.isNotEmpty(sortField) && order != null) {  
            reqBuilder.addSort(sortField, order);  
        }  
        if (from >= 0 && size > 0) {  
            reqBuilder.setFrom(from).setSize(size);  
        }  
        SearchResponse resp = reqBuilder.execute().actionGet();  
        SearchHit[] hits = resp.getHits().getHits();  
        ArrayList<String> results = new ArrayList<String>();  
        for (SearchHit hit : hits) {  
            results.add(hit.getId());  
        }  
        return results;  
    }  
    
    /**
     * 获取某个字段总和
     * @param type
     * @param field
     * @param start
     * @param end
     * @return
     */
    public long dataCount(String type, String field, long start, long end, String url){
        TransportClient client = getClient();
        SumBuilder visitTimesSumAgg= AggregationBuilders.sum("sumAgg").field(field); 
        //url = QueryParser.escape(url);
        SearchResponse searchResponse = client.prepareSearch(esIndexName)  
            .setTypes(type)  
            .setQuery(QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("date").gte(start).lte(end)).must(QueryBuilders.wildcardQuery("url", url)))
            .addAggregation(visitTimesSumAgg)
            //.setQuery(QueryBuilders.matchAllQuery())
            .setSearchType(SearchType.QUERY_THEN_FETCH)
            .get();
        Map<String, Aggregation> aggMap = searchResponse.getAggregations().asMap();
        long sumVisitTimesTerms = (long)((InternalSum) aggMap.get("sumAgg")).getValue();
        /*SearchHits hits = searchResponse.getHits();  
        int total = (int)hits.getTotalHits();*/
        client.close();
        return sumVisitTimesTerms;
    }
    
    /**
     * 查询分组后的数据
     * @param type
     * @param sortField
     * @param order
     * @param start
     * @param end
     * @return
     */
    public List<Map<String, Object>> AggregationData(String type, String sortField, SortOrder order, long start, long end, String url) {  
        TransportClient client = getClient();
        TermsBuilder provinceAgg= AggregationBuilders.terms("province_team").field("province").size(0);  
        TermsBuilder cityAgg= AggregationBuilders.terms("city_team").field("city").size(0);
        TermsBuilder operatorsAgg= AggregationBuilders.terms("operators_team").field("operators").size(0);  
        TermsBuilder networkTypeAgg= AggregationBuilders.terms("network_type_team").field("network_type").size(0);
        SumBuilder loadTimeSumAgg= AggregationBuilders.sum("sum_total_load_time").field("total_load_time");
        SumBuilder visitTimesSumAgg= AggregationBuilders.sum("sum_visit_times").field("visit_times");
        //url = QueryParser.escape(url);
        
        SearchResponse searchResponse = client.prepareSearch(esIndexName)  
            .setTypes(type)
            .setSearchType(SearchType.QUERY_THEN_FETCH)  
            .setQuery(QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("date").gte(start).lte(end)).must(QueryBuilders.wildcardQuery("url", url)))
            .addAggregation(provinceAgg.subAggregation(cityAgg.subAggregation(operatorsAgg.subAggregation(networkTypeAgg.subAggregation(loadTimeSumAgg)
                .subAggregation(visitTimesSumAgg)))))
            //QueryBuilders.prefixQuery("url", url)
            //.setQuery(QueryBuilders.rangeQuery("date").gte(start).lte(end))
            //.setQuery(QueryBuilders.matchAllQuery())
            //.setQuery(QueryBuilders.prefixQuery("url", "http")) //查询所有  
            //.setQuery(QueryBuilders.matchQuery("name", "tom").operator(Operator.AND)) //根据tom分词查询name,默认or  
            //.setQuery(QueryBuilders.multiMatchQuery("tom", "name", "age")) //指定查询的字段  
            //.setQuery(QueryBuilders.queryString("name:to* AND age:[0 TO 19]")) //根据条件查询,支持通配符大于等于0小于等于19  
            //.setQuery(QueryBuilders.termQuery("name", "tom"))//查询时不分词  
            //.setFrom(from).setSize(size)//分页  
            .addSort(sortField, SortOrder.DESC)//排序  
            .get();  
        
        List<Map<String, Object>> results = new ArrayList<Map<String, Object>>();
        Map<String, Aggregation> provinceAggMap = searchResponse.getAggregations().asMap();  
        StringTerms province_teamAgg= (StringTerms) provinceAggMap.get("province_team");
        Iterator<Bucket> provinceTeamBucketIt = province_teamAgg.getBuckets().iterator();  
        while (provinceTeamBucketIt .hasNext()) {  
            Bucket provinceBuck = provinceTeamBucketIt .next();  
            //省份  
            String province = provinceBuck.getKey().toString();  
            //得到子聚合  
            Map<String, Aggregation> cityAggMap = provinceBuck.getAggregations().asMap();  
            StringTerms city_termAgg = (StringTerms) cityAggMap.get("city_team");
            Iterator<Bucket> cityTeamBucketIt = city_termAgg.getBuckets().iterator();
            while (cityTeamBucketIt .hasNext()) { 
                Bucket cityBuck = cityTeamBucketIt .next();
                String city = cityBuck.getKey().toString();
                Map<String, Aggregation> operatorsAggMap = cityBuck.getAggregations().asMap();
                StringTerms operators_termAgg = (StringTerms) operatorsAggMap.get("operators_team");
                Iterator<Bucket> operatorsTeamBucketIt = operators_termAgg.getBuckets().iterator();
                while (operatorsTeamBucketIt .hasNext()) { 
                    Bucket operatorsBuck = operatorsTeamBucketIt .next();
                    String operators = operatorsBuck.getKey().toString();
                    Map<String, Aggregation> networkTypeAggMap = operatorsBuck.getAggregations().asMap();
                    StringTerms network_type_termAgg = (StringTerms) networkTypeAggMap.get("network_type_team");
                    Iterator<Bucket> networkTypeTeamBucketIt = network_type_termAgg.getBuckets().iterator();
                    while (networkTypeTeamBucketIt .hasNext()) { 
                        Bucket networkTypeBuck = networkTypeTeamBucketIt .next();
                        String networkType = networkTypeBuck.getKey().toString();
                        Map<String, Aggregation> subAggMap = networkTypeBuck.getAggregations().asMap();
                        double sumTotalLoadTimeTerms =  ((InternalSum) subAggMap.get("sum_total_load_time")).getValue();
                        long sumVisitTimesTerms = (long)((InternalSum) subAggMap.get("sum_visit_times")).getValue();
                        Map<String, Object> map = new HashMap<String, Object>();
                        map.put("province", province);
                        map.put("city", city);
                        map.put("operators", operators);
                        map.put("network_type", networkType);
                        map.put("total_load_time", sumTotalLoadTimeTerms);
                        map.put("visit_times", sumVisitTimesTerms);
                        results.add(map);
                    }
                }
            }
        }
        
        /*SearchHits hits = searchResponse.getHits();  
        long total = hits.getTotalHits();  
        SearchHit[] searchHits = hits.hits(); 
        List<Map<String, Object>> results = new ArrayList<Map<String, Object>>();
        for(SearchHit s : searchHits)  
        {  
            results.add(s.getSource());
        } */
        client.close();
        return results;  
    }  
  
  
    /**  
     * QueryBuilders 所有查询入口  
     */  
    public List<Map<String, Object>> queryForObjectEq(String type,  
            String[] fields, String content, String sortField, SortOrder order,  
            int from, int size) {  
        TransportClient client = getClient();
        SearchRequestBuilder reqBuilder = client.prepareSearch(esIndexName)  
                .setTypes(type).setSearchType(SearchType.DEFAULT)  
                .setExplain(true);  
        QueryStringQueryBuilder queryString = QueryBuilders.queryStringQuery("\""  
                + content + "\"");  
        for (String k : fields) {  
            queryString.field(k);  
        }  
        queryString.minimumShouldMatch("10");  
        reqBuilder.setQuery(QueryBuilders.boolQuery().must(queryString))  
                .setExplain(true);  
        if (StringUtils.isNotEmpty(sortField) && order != null) {  
            reqBuilder.addSort(sortField, order);  
        }  
        if (from >= 0 && size > 0) {  
            reqBuilder.setFrom(from).setSize(size);  
        }  
  
        SearchResponse resp = reqBuilder.execute().actionGet();  
        SearchHit[] hits = resp.getHits().getHits();  
  
        List<Map<String, Object>> results = new ArrayList<Map<String, Object>>();  
        for (SearchHit hit : hits) {  
            results.add(hit.getSource());  
        }  
        client.close();
        return results;  
    }  
  
    public void afterPropertiesSet() throws Exception {  
        System.out.println("init...");  
  
    }  
  
}  
